package com.ysstech.kettle.service.impl;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Future;

import com.ysstech.kettle.entity.JobParam;
import com.ysstech.kettle.init.KettleConfig;
import com.ysstech.kettle.service.JobExecuteService;
import com.ysstech.kettle.service.JobRecordService;
import com.ysstech.kettle.service.JobRelationService;
import com.ysstech.kettle.util.ThreadPool;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.pentaho.di.core.Props;
import org.pentaho.di.core.gui.HasOverwritePrompter;
import org.pentaho.di.core.gui.OverwritePrompter;
import org.pentaho.di.core.logging.LogChannelInterface;
import org.pentaho.di.i18n.BaseMessages;
import org.pentaho.di.job.JobMeta;
import org.pentaho.di.repository.*;
import org.pentaho.di.trans.TransMeta;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;


/**
 * <p>
 * 作业表 服务实现类
 * </p>
 *
 * @author lishuangliang
 * @since 2021-06-03
 */
@Service
@Slf4j
public class JobExecuteExecuteServiceImpl implements JobExecuteService {

    @Autowired
    private JobRecordService jobRecordService;
    @Autowired
    private KettleConfig kettleConfig;
    @Autowired
    private JobRelationService jobRelationService;

    /**
     * 导出资源库文件信息
     *
     * @param outputPath
     * @param xmlFileName
     * @param jobPath
     * @throws Exception D:\table\
     */
    public void exportTemplateXml(String outputPath, String xmlFileName, String jobPath) throws Exception {
        Repository repository = kettleConfig.getKettleRepository();
        try {
            IRepositoryExporter exporter = repository.getExporter();
            exporter.exportAllObjects(null, outputPath + xmlFileName,
                    StringUtils.isEmpty(jobPath) ? repository.loadRepositoryDirectoryTree() :
                            repository.findDirectory(jobPath), RepositoryExporter.ExportType.ALL.name());
        } catch (Exception e) {
            log.error("导出kettle配置失败", e);
        } finally {
            repository.disconnect();
        }
        return;
    }

    /**
     * 导入资源库文件信息
     *
     * @param filePath
     * @param fileName
     * @throws Exception
     */
    public void importTemplateXml(String filePath, String fileName) throws Exception {
        Repository repository = kettleConfig.getKettleRepository();
        try {
            boolean initialized = Props.isInitialized();
            if (!initialized) {
                Props.init(1);
            }
            IRepositoryImporter importer = repository.getImporter();
            List<String> filenames = new ArrayList<>();
            filenames.add(fileName);
            RepositoryImportFeedbackInterface feedbackInterface = new ImportFeedback(
                    repository.getLog(), false, true,
                    new BufferedReader(new InputStreamReader(System.in)));
            importer.importAll(feedbackInterface, filePath,
                    filenames.toArray(new String[0]),
                    repository.loadRepositoryDirectoryTree(), true, false, "true");
        } catch (Exception e) {
            log.error("导入kettle配置失败", e);
        } finally {
            repository.disconnect();
        }
    }

    /**
     * @param ids
     * @param isTaskDetail 是否执行的明细任务
     * @param isTask       是否执行任务
     * @throws Exception
     */

    public void execute(List<String> ids, List<JobParam> listParam, boolean isTaskDetail, boolean isTask) throws Exception {
        this.doExecute(ids, listParam, isTaskDetail, isTask, null);
    }

    /**
     * 递归执行 ，依赖关系的先执行被依赖的成功后在执行
     *
     * @param ids
     * @param listParam
     * @param isTaskDetail
     * @param isTask
     * @param map
     * @throws Exception
     */
    public void doExecute(List<String> ids, List<JobParam> listParam, boolean isTaskDetail, boolean isTask, Map<String, Boolean> map) throws Exception {
        if (null == map) {
            map = new HashMap<>();
        }
        // 获取依赖关系
        Map<String, List<String>> mapRelation = null;
        if (!CollectionUtils.isEmpty(ids) && ids.size() > 1) {
            mapRelation = jobRelationService.queryRelation(ids);
        }
        List<String> idsChildren = new ArrayList<>();
        Map<String, Future<Boolean>> mapFuture = new HashMap<>();
        falgLoop:
        for (String taskId : ids) {
            if (StringUtils.isBlank(taskId)) {
                continue;
            }
            if (!CollectionUtils.isEmpty(mapRelation) && mapRelation.containsKey(taskId)) {
                List<String> list = mapRelation.get(taskId);
                if (!CollectionUtils.isEmpty(list)) {
                    for (String relationId : list) { //循环每个依赖的任务是不是已经执行完成
                        if (null == map.get(relationId) || !map.get(relationId)) {
                            idsChildren.add(taskId);  // 把还有依赖没有执行完成的数据，访问递归你们，下次在执行
                            continue falgLoop; //循环下一个
                        }
                    }
                }
            }
            KtttleExecuteThread ht = new KtttleExecuteThread(jobRecordService, kettleConfig, listParam, isTaskDetail, isTask, taskId);
            Future<Boolean> future = ThreadPool.getThreadPool().submit(ht);
            mapFuture.put(taskId, future);
        }
        for (Map.Entry entry: mapFuture.entrySet()) {
            Future<Boolean> future=(Future<Boolean>)entry.getValue();
            map.put(entry.getKey().toString(),future.get()); //等到任务执行完成
        }
        if (!CollectionUtils.isEmpty(idsChildren)) { //递归到所有的任务都执行完成
            doExecute(idsChildren, listParam, isTaskDetail, isTask, map);
        }
    }

    private static final Class<?> PKG = ImportFeedback.class;

    private static class ImportFeedback implements RepositoryImportFeedbackInterface, HasOverwritePrompter {
        private final LogChannelInterface log;
        private final boolean continueOnError;
        private final boolean replace;
        private final BufferedReader reader;

        public ImportFeedback(LogChannelInterface log, boolean continueOnError, boolean replace, BufferedReader reader) {
            this.log = log;
            this.continueOnError = continueOnError;
            this.replace = replace;
            this.reader = reader;
        }

        public void updateDisplay() {
        }

        public boolean transOverwritePrompt(TransMeta transMeta) {
            return this.replace;
        }

        public void showError(String title, String message, Exception e) {
            this.log.logError(title + " : " + message, e);
        }

        public void setLabel(String labelText) {
            this.log.logBasic(labelText);
        }

        public boolean jobOverwritePrompt(JobMeta jobMeta) {
            return this.replace;
        }

        public boolean askContinueOnErrorQuestion(String title, String message) {
            return this.continueOnError;
        }

        public void addLog(String line) {
            this.log.logBasic(line);
        }

        public boolean isAskingOverwriteConfirmation() {
            return false;
        }

        public OverwritePrompter getOverwritePrompter() {
            return new OverwritePrompter() {
                private final String yes;
                private final String no;
                private final String none;
                private final String all;
                private final String prompt;

                {
                    this.yes = BaseMessages.getString(PKG, "Import.Yes", new String[0]);
                    this.no = BaseMessages.getString(PKG, "Import.No", new String[0]);
                    this.none = BaseMessages.getString(PKG, "Import.None", new String[0]);
                    this.all = BaseMessages.getString(PKG, "Import.All", new String[0]);
                    this.prompt = "[" + this.yes + "," + this.no + "," + this.none + "," + this.all + "]";
                }

                public boolean overwritePrompt(String message, String rememberText, String rememberPropertyName) {
                    ImportFeedback.this.log.logBasic(message);
                    Boolean result = null;
                    boolean remember = false;

                    while (result == null) {
                        ImportFeedback.this.log.logBasic(this.prompt);

                        String line;
                        try {
                            line = ImportFeedback.this.reader.readLine().trim();
                        } catch (IOException var8) {
                            throw new RuntimeException(BaseMessages.getString(PKG, "Import.CouldntReadline", new String[0]));
                        }

                        if (!line.equalsIgnoreCase(this.yes) && !line.equalsIgnoreCase(this.all)) {
                            if (line.equalsIgnoreCase(this.no) || line.equalsIgnoreCase(this.none)) {
                                result = false;
                            }
                        } else {
                            result = true;
                        }

                        if (line.equalsIgnoreCase(this.all) || line.equalsIgnoreCase(this.none)) {
                            remember = true;
                        }
                    }

                    Props.getInstance().setProperty(rememberPropertyName, !remember ? "Y" : "N");
                    return result;
                }
            };
        }
    }
}
